package com.titan.single.kafka;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.titan.single.constant.KafkaConstant;
import com.titan.single.entity.User;
import com.titan.single.util.JsonUtils;
import com.titan.single.vo.CanalCommon;
import com.titan.single.vo.KafkaMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

/**
 * canal消息 消费者
 */
@Service
@Slf4j
public class CanalConsumer {

    @Autowired
    private ObjectMapper objectMapper;

    @KafkaListener(topics = KafkaConstant.CANAL_TOPIC, groupId = KafkaConstant.GROUP_TIAN)
    public void consumeMessageCanal(@Payload String message,
                                    @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                    @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                                    @Header(KafkaHeaders.OFFSET) long offset,
                                    Acknowledgment acknowledgment) {
        try {

            CanalCommon<User> canalCommon = JsonUtils.stringToObjName(message, new TypeReference<CanalCommon<User>>() {});
            log.info("接受消息对象:"+JsonUtils.objToString(canalCommon));
            // 处理业务逻辑
            processMessage(message);

            // 手动提交偏移量
            acknowledgment.acknowledge();
            log.info("消费者---消息处理完成并提交偏移量");

        } catch (Exception e) {
            log.error("消费者---消费消息异常: {}", e.getMessage(), e);
            // 根据业务需要决定是否提交偏移量
        }
    }



    private void processMessage(String message) {
        // 在这里添加你的业务处理逻辑

        // 模拟业务处理
        try {
            Thread.sleep(100); // 模拟处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("消息处理被中断: {}", message);
        }
    }
}